package com.springboot.rabbitmq.config;

import com.springboot.rabbitmq.queue.QueueEnum;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ReturnListener;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.io.IOException;
import java.util.List;

@Configuration
@Slf4j
@Data
public class RabbitConfig {

    @Value("${spring.rabbitmq.host:test}")
    private String host;

    @Value("${spring.rabbitmq.port:test}")
    private String port;

    @Value("${spring.rabbitmq.username:test}")
    private String username;

    @Value("${spring.rabbitmq.password:test}")
    private String password;

    @Value("${spring.rabbitmq.virtual-host:test}")
    private String virtualHost;

    @Value("${spring.rabbitmq.publisher-confirm-type:NONE}")
    private String publisherConfirms;

    @Value("${spring.rabbitmq.publisher-returns:false}")
    private boolean publisherReturns;

    @Value("${spring.rabbitmq.connection-timeout:test}")
    private int connectionTimeOut;

    @Value("${spring.application.name:test}")
    private String appName;

    @Value("${spring.rabbitmq.template.mandatory:false}")
    private boolean mandatory;

    private final static String DIRECT_EXCHANGE = "direct_exchange";
    private final static String DIRECT_ROUTINGKEY = "direct_routing";


    @Bean
    public CachingConnectionFactory connectionFactory() {

        CachingConnectionFactory connectionFactory = new CachingConnectionFactory();
        connectionFactory.setAddresses(host + ":" + port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualHost);
        /** 如果要进行消息回调，则这里必须要设置为true */
        if (publisherConfirms.toUpperCase().equals(CachingConnectionFactory.ConfirmType.SIMPLE.toString())) {
            connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.SIMPLE);
        } else if (publisherConfirms.toUpperCase().equals(CachingConnectionFactory.ConfirmType.CORRELATED.toString())) {
            connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        } else {
            connectionFactory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.NONE);
        }
        connectionFactory.setPublisherReturns(publisherReturns);
        connectionFactory.setConnectionTimeout(connectionTimeOut);
        try {
            mqMsgQueue(connectionFactory);
        }catch (Exception e){
            e.printStackTrace();
            log.error("绑定rabbitmq失败",e);
        }

        return connectionFactory;
    }

    @Bean
    public AmqpAdmin amqpAdmin() {
        return new RabbitAdmin(connectionFactory());
    }

    /**
     定义rabbit template用于数据的接收和发送
     **/
    @Bean
    public RabbitTemplate rabbitTemplate() {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory());
        /**若使用confirm-callback或return-callback，必须要配置publisherConfirms或publisherReturns为true
         * 每个rabbitTemplate只能有一个confirm-callback和return-callback*/
        rabbitTemplate.setConfirmCallback(new MsgSendConfirmCallBack());
        rabbitTemplate.setReturnsCallback(new MsgSendReturnCallback());
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        /**
         * 使用return-callback时必须设置mandatory为true，或者在配置中设置mandatory-expression的值为true，可针对每次请求的消息去确定’mandatory’的boolean值，
         * 只能在提供’return -callback’时使用，与mandatory互斥*/
        rabbitTemplate.setMandatory(mandatory);

        return rabbitTemplate;
    }




    public void mqMsgQueue(CachingConnectionFactory connectionFactory) throws AmqpException, IOException {

        /**判断是否有需要创建交换机的模块*/
        List<String[]> queueNameList = QueueEnum.getQueueByModelExchange(appName);
        if(null==queueNameList || queueNameList.isEmpty()){
            log.info("判断是否有需要创建交换机的模块为空，appName：{}", appName);
            return;
        }



        /**创建交换机*/
        Connection connection = connectionFactory.createConnection();
        Channel channel = connection.createChannel(false);
        channel.exchangeDeclare(appName, BuiltinExchangeType.DIRECT);

        for(String[] queueNames : queueNameList){
            this.binding(channel, appName, queueNames[1], queueNames[2], queueNames[3], queueNames[0]);
        }

        channel.addReturnListener(new ReturnListener() {

            @Override
            public void handleReturn(int arg0, String arg1, String arg2, String arg3, AMQP.BasicProperties arg4, byte[] arg5)
                    throws IOException {
                //此处便是执行Basic.Return之后回调的地方
                String message = new String(arg5);
                System.out.println("Basic.Return返回的结果:  "+message);
            }
        });

    }



    /**
     *
     * @author zyj
     * @date 2019/3/8 16:25
     * @param exchangeName 交换机名称
     * @param routingkey
     * @param queueName
     * @param descrip 描述
     * @param bindName 绑定的服务名
     * @return
     */
    public void binding(Channel channel, String exchangeName, String routingkey, String queueName, String descrip, String bindName) throws AmqpException, IOException{
        log.info(bindName+"绑定rabbitmq：["+routingkey+":"+queueName+":"+descrip+"]");
        channel.queueDeclare(queueName, true, false, false, null);
        channel.queueBind(queueName, exchangeName, routingkey);
    }





    @Bean
    public RabbitAdmin rabbitTemplateAdmin() {
        return new RabbitAdmin(connectionFactory());
    }



}
